/*
 * Update.cpp
 *
 *  Created on: Aug 29, 2016
 *      Author: ubuntu
 */

#include "Update.h"
#include "common.h"
#include "Conf.h"
#include <syslog.h>
#include <stdio.h>
#include <errno.h>
#include "DBPool.h"

#include <sys/epoll.h>
#include <unistd.h>
#include <sys/types.h>
#include <netdb.h>
#include <netinet/in.h>
#include <linux/un.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <fcntl.h>

#include "N900/N900Resolve.h"

#include "G2O/G2OResolve.h"

Update::Update()
{
	m_bTerminated = false;
	m_pDBPool = NULL;
	m_fd = -1;
	m_epfd = -1;
	unix_path = NULL;
	m_n900rsv = NULL;
	m_g2orsv = NULL;
}

Update::~Update()
{
	if (unix_path)
		unlink(unix_path);
	if (m_n900rsv)
		delete m_n900rsv;
	if (m_g2orsv)
		delete m_g2orsv;
}

bool Update::init()
{
	m_pDBPool = getConf()->cpool;
	if (!m_pDBPool)
		return false;
	return true;
}

void Update::start()
{
	if (dminit())
		dmrun();
}

void Update::stop()
{
	m_bTerminated = true;
	if (m_n900rsv)
		m_n900rsv->stop();
	if (m_g2orsv)
		m_g2orsv->stop();
}

void *Update::DMThread(void *arg)
{
	Update *pUpdate = (Update *)arg;
	if (pUpdate->dminit())
		pUpdate->dmrun();
	return NULL;
}

void *Update::NRsolveThread(void *arg)
{
	N900Resolve *pResolve = (N900Resolve *)arg;
	pResolve->run();
	return NULL;
}

void *Update::GRsolveThread(void *arg)
{
	G2OResolve *pResolve = (G2OResolve *)arg;
	pResolve->run();
	return NULL;
}

bool Update::dminit()
{
	m_n900rsv = new N900Resolve();
	m_g2orsv = new G2OResolve();
	m_g2orsv->init();
	unix_path = getConf()->unix_path;
	if (unix_path == NULL)
		unix_path = strdup("net/socket");
	struct epoll_event event;
	struct sockaddr_un addr;
	m_epfd = epoll_create(32000);
	if (m_epfd < 0)
		return false;
	m_fd = socket(AF_UNIX, SOCK_STREAM, 0);
	if (m_fd < 0) {
		close(m_epfd);
		return false;
	}

	unlink(unix_path);
	bzero(&addr, sizeof(addr));
	addr.sun_family = AF_UNIX;
	strncpy(addr.sun_path, unix_path, sizeof(addr.sun_path)-1);

	int on = 1;
	setnonblocking(m_fd);
	setsockopt(m_fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
	if (bind(m_fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
		close(m_fd);
		close(m_epfd);
		return false;
	}

	if (listen(m_fd, MAX_EVENTS) < 0)
	{
		close(m_fd);
		close(m_epfd);
		return false;
	}

	event.events = EPOLLIN;
	event.data.fd = m_fd;
	epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_fd, &event);
	return true;
}

void Update::dmrun()
{
	pthread_t tid;
	for (int i = 0; i < 6; i++) {
		pthread_create(&tid, NULL, Update::NRsolveThread, (void *)m_n900rsv);
		pthread_detach(tid);
	}
	for (int i = 0; i < 6; i++) {
		pthread_create(&tid,  NULL, Update::GRsolveThread, (void *)m_g2orsv);
		pthread_detach(tid);
	}
	int res, nevent = NEVENTS;
	struct epoll_event event, *wait_events;
	wait_events = (struct epoll_event *)calloc(nevent, sizeof(epoll_event));
	if (!wait_events)
		stop();

	while (!m_bTerminated) {
		res = epoll_wait(m_epfd, wait_events, nevent, 10);
		if (res < 0) {
			if (errno != EINTR) {
				stop();
				return;
			}
			continue;
		}
		if (res == 0)
			continue;
		for (int i = 0; i < res; i++) {
			int kfd = wait_events[i].data.fd;
			if (kfd == m_fd) {
re_accept:
				int fd = accept(m_fd, NULL, NULL);
				if (fd < 0) {
					debug(LOG_DEBUG, "accept: m_fd: %d %d [%s]", m_fd, fd, strerror(errno));
					if (errno != EINTR) {
						stop();
						return;
					}
					goto re_accept;
				}
				setnonblocking(fd);
				event.events = EPOLLIN|EPOLLET;
				event.data.fd = fd;
				epoll_ctl(m_epfd, EPOLL_CTL_ADD, fd, &event);
				continue;
			}
			dmread(kfd);
		}

		if (res == nevent && res < MAX_EVENTS) {
			int new_nevent = nevent * 2;
			struct epoll_event *new_events = (struct epoll_event *)realloc(wait_events, new_nevent * sizeof(struct epoll_event));
			if (new_events) {
				debug(LOG_DEBUG, "new_nevent: %d", new_nevent);
				nevent = new_nevent;
				wait_events = new_events;
			}
		}
	}
}

void Update::dmread(int fd)
{
	int to_alloc = MIN_MALLOC_SIZE;
	unsigned char *buf = (unsigned char *)malloc(to_alloc);
	if (!buf) {
		stop();
		return;
	}
	int n = 0, nread;
	while (1) {
		if (n >= to_alloc) {
			to_alloc *= 2;
			unsigned char *tmp = (unsigned char *)realloc(buf, to_alloc);
			if (tmp)
				buf = tmp;
			else
				exit(0);
		}
		nread = read(fd, buf+n, to_alloc-n);
		if (nread < 0) {
			if (errno != EAGAIN) {
				debug(LOG_DEBUG, "Read error: %d[%s]", n, strerror(errno));
				close(fd);
				epoll_ctl(m_epfd, EPOLL_CTL_DEL, fd, NULL);
				break;
			}
			break;
		} else if (nread == 0) {
			close(fd);
			epoll_ctl(m_epfd, EPOLL_CTL_DEL, fd, NULL);
			break;
		}
		n += nread;
	}

	if (n > 0) {
		dmparse(buf, n);
	} else {
		if (buf)
			free(buf);
	}
}

void Update::dmparse(unsigned char *buf, size_t len)
{
	if (memcmp(buf, DM_TAG_HEAD, DM_TAG_HEAD_LEN) != 0)
	{
		free(buf);
		debug(LOG_DEBUG, "错误消息");
		return;
	}
	unsigned char *p = buf + DM_TAG_HEAD_LEN;
	unsigned char type = p[0];
	switch (type) {
	case SV_CMD:
		free(buf);
		break;
	case SV_G2O:
		m_g2orsv->getBuf(buf, len);
		break;
	case SV_N900:
		m_n900rsv->getBuf(buf, len);
		break;
	case SV_GOLO:
		free(buf);
		break;
	default:
		debug(LOG_DEBUG, "Unknow type: %d", type);
		free(buf);
		break;
	}
}
